#include "TCPServlet.h"
#include <stdio.h> /* for LogI() */
#include <sys/types.h> /* for Socket data types */
#include <sys/socket.h> /* for socket(), connect(), send(), and recv() */
#include <arpa/inet.h>
#include <sys/un.h> /* for sockaddr_un */
#include <netinet/in.h> /* for IP Socket data types */
#include <stdlib.h> /* for exit() */
#include <string.h> /* for memset() */
#include <unistd.h> /* for close() */
#include <errno.h> /* for errno*/
#include <thread>
#include <time.h>
#include <sys/time.h>
#include "ILogger.h"

class TCPSocketSession : public ISession {
public:
    int socket;
    TCPSocketSession(int socket) :socket(socket){

    }
    void reply(const void *data,int len)
    {
        send(socket,data,len,0);
    }
    void close(){

    }
};

void TCPClientServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler) {
    msgidHandlers[type].push_back( handler);
}
void TCPClientServlet::request(const void *data,int len) {
    int i = send(conn_fd, data, len, 0);
    LogI("TCPClientServlet::request want:%d real:%d\n",len,i);
}
void TCPClientServlet::start() {
     auto clientConnectionHandlerThread = new std::thread(&TCPClientServlet::threadLogic, this);
     _thread = clientConnectionHandlerThread;
}
void TCPClientServlet::wait() {
    ((std::thread*)_thread)->join();
}
void TCPClientServlet::stop() {
    isRun = false;
}
void TCPClientServlet::onLoad(INode *node) {
    INode *child = node->childNode("connection");
    if (child) {
        ip = child->getStr("ip");
        port = child->get<int>("port");
    }
    start();
}
void TCPClientServlet::onUnload() {
   // stop();
    wait();
}
void TCPClientServlet::onMessage(ISession *session,void* data,int size) {
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void TCPClientServlet::onState(SERVLET_STATUS status,ISession *session,int code) {
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
      for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}
TCPClientServlet::~TCPClientServlet() {

}
void TCPClientServlet::recvthreadLogic(int socket) {
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    TCPSocketSession session(socket);
    onState(OPEN,&session,0);
    while (isRun) {
        char buf[1024];
        memset(buf, 0, 1024);
        int n = recv(socket, buf, 1024, 0);
         if(n == -1) { //recv函数返回错误

            if (errno == EAGAIN) {
                continue;
            }
            LogI("client recv error\n");
            onState(ERROR,&session,0);
            close(socket);
            socket = -1;
            break;
        } else if(n == 0) { //client端套接字被关闭
            LogI("xx client is closed");
            onState(CLOSE,&session,0);
            close(socket);
            socket = -1;
            break;
        } else { //读取成功
            /*对收到的数据进行处理*/
            LogI("client receive dataLen:%d", n);
           
            onMessage(&session,buf,n);
        }
    }
    LogI("TCPClientServlet::recvthreadLogic end");
}

void TCPClientServlet::threadLogic() {

    struct sockaddr_in  servaddr;

    if( (conn_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0){
        LogI("create socket error: %s(errno: %d)", strerror(errno),errno);
        return;
    }

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    if( inet_pton(AF_INET,ip.c_str(), &servaddr.sin_addr) <= 0){
        LogI("inet_pton error for %s:%d",ip.c_str(),port);
        return;
    }

    int retryCount = 0;
    while(connect(conn_fd, (struct sockaddr *)&servaddr, sizeof(servaddr)) == -1 && retryCount++ < 3)
    {
        LogI("connect error retryCount:%d",retryCount);
        std::this_thread::sleep_for(std::chrono::milliseconds(1500));
    }
    LogI("connect successfully!");
    auto client = std::thread(&TCPClientServlet::recvthreadLogic,this,conn_fd);
    client.join();
    close(conn_fd);
    return;
}

void TCPServerServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler) {
    msgidHandlers[type].push_back(handler);
}
void TCPServerServlet::request(const void *data,int len) {

}
void TCPServerServlet::start() {

    auto clientConnectionHandlerThread = new std::thread(&TCPServerServlet::threadLogic, this);
  
    _thread = clientConnectionHandlerThread;
}
void TCPServerServlet::wait() {

}
void TCPServerServlet::stop() {

}
void TCPServerServlet::onLoad(INode *node) {
    INode *child = node->childNode("connection");
    if (child) {
        ip = child->getStr("ip");
        port = child->get<int>("port");
    }
    start();
}
void TCPServerServlet::onUnload() {
    (( std::thread *)_thread)->join();
}
void TCPServerServlet::onMessage(ISession *session,void* data,int size) {
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void TCPServerServlet::onState(SERVLET_STATUS status,ISession *session,int code) {
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}
void TCPServerServlet::recvthreadLogic(int socket) {
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    TCPSocketSession session(socket);
    onState(OPEN,&session,0);
    while (isRun) {
        char buf[1024];
        memset(buf, 0, 1024);
        int n = recv(socket, buf, 1024, 0);
         if(n == -1) { //recv函数返回错误

            if (errno == EAGAIN) {
                continue;
            }
            LogI("server recv err!\n");
            onState(ERROR,&session,0);
            close(socket);
            socket = -1;
            break;
        } else if(n == 0) { //client端套接字被关闭
            LogI("in server ,client is closed\n");
            onState(CLOSE,&session,0);
            close(socket);
            socket = -1;
            break;
        } else { //读取成功
            /*对收到的数据进行处理*/
            LogI("server receive data:%d", n);
           // send(socket,buf,n,0);
            
            onMessage(&session,buf,n);
        }
    }
    LogI("DomainServerServlet::recvThreadLogic end");
}
TCPServerServlet::~TCPServerServlet() {

}

void TCPServerServlet::threadLogic() {
    int socketfd = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0);
    if (socketfd == -1)
    {
        LogI("Error initializing socket %d", errno);
        return;
    }
    int optionValue = 0;
    if ((setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR, (char*) &optionValue, sizeof(optionValue))
        != 0)
        || (setsockopt(socketfd, SOL_SOCKET, SO_KEEPALIVE, (char*) &optionValue, sizeof(optionValue))
            != 0))
    {
        LogI("Error setting options %d", errno);
        return;
    }
    sockaddr_in agentAddress = {0};


    agentAddress.sin_family = AF_INET;
    agentAddress.sin_port = htons(port);
    memset(&(agentAddress.sin_zero), 0, 8);
    agentAddress.sin_addr.s_addr = INADDR_ANY;

    if (bind(socketfd, (sockaddr*) &agentAddress, sizeof(agentAddress)) != 0)
    {
        LogI( "Error binding to socket, make sure nothing else is listening on this port %d", errno);
        return ;
    }

    if (listen(socketfd, 100) != 0)
    {
        LogI("Error listening %d", errno);
        return ;
    }
    std::vector<std::thread*> clients;
    std::vector<std::thread*> threads;
    struct timeval tv_out;
    tv_out.tv_sec = 3;
    tv_out.tv_usec = 0;
    setsockopt(socketfd, SOL_SOCKET, SO_RCVTIMEO, &tv_out, sizeof(tv_out));
    while (isRun)
    {
        sockaddr_in clientAddress;

        socklen_t addressLength = {0};
        int socketClient = -1;
        try {
            socketClient = accept((socketfd), (sockaddr*) &(clientAddress), &(addressLength));
            if (socketClient != -1)
            {
                auto client = new std::thread(&TCPServerServlet::recvthreadLogic,this,socketClient);
                clients.push_back(client);
            }
        } catch(...) {
            LogI("new thread failed");
            if (socketClient >= 0) {
                close(socketClient);
            }
        }

    }

    if (socketfd != -1)
    {
        close(socketfd);
    }
    for (auto client = clients.begin(); client != clients.end(); ++client)
    {
        if (*client)
            (*client)->join();
    }
    LogI("server exit");
}


REG_PLUGIN(TCPServerServlet);
REG_PLUGIN(TCPClientServlet);
